package ar.uba.fi.tonyvaliente.tree;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;

import ar.uba.fi.tonyvaliente.files.Persistor;
import ar.uba.fi.tonyvaliente.files.PersistorException;

public class FileBPlusTreeNodePersistor<K extends Comparable<K>, V> extends Persistor<BPlusTreeNode<K, V>> {
	private FileBPlusTree<K, V> tree;
	
	public FileBPlusTreeNodePersistor(FileBPlusTree<K, V> tree)	{
		this.setTree(tree);
	}
	
	@Override
	public byte[] serialize(BPlusTreeNode<K, V> node) throws PersistorException {
		if (node instanceof InnerBPlusTreeNode) {
			return serializeInnerNode((InnerBPlusTreeNode<K, V>)node);
		} else if (node instanceof LeafBPlusTreeNode) {
			return serializeLeafNode((LeafBPlusTreeNode<K, V>)node);
		} else {
			// TODO: wrong node kind, change exception message
			throw new PersistorException("1");
		}
	}

	@Override
	public BPlusTreeNode<K, V> deserialize(byte[] bytes) throws PersistorException {
		if (bytes[0] == 1) {
			return deserializeInnerNode(bytes);
		} else if (bytes[0] == 0) {
			return deserializeLeafNode(bytes);
		} else {
			// TODO: wrong bytes, change exception's message
			throw new PersistorException("1");
		}
	}

	private byte[] serializeInnerNode(InnerBPlusTreeNode<K, V> node) throws PersistorException {
		try {
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			DataOutputStream output = new DataOutputStream(baos);
			
			output.writeByte(1);

			writeFileReference(output, (FileBPlusTreeNodeReference<K, V>)node.getParentReference());

			List<K> keys = node.getKeys();
			output.writeInt(keys.size());
			for (K key : keys) {
				writeKey(output, key);
			}
			
			List<BPlusTreeNodeReference<K, V>> references = node.getReferences();
			for (BPlusTreeNodeReference<K, V> reference : references) {
				writeFileReference(output, (FileBPlusTreeNodeReference<K, V>)reference);
			}
			
			output.close();
			return baos.toByteArray();
		} catch (IOException e) {
			e.printStackTrace();
			// TODO: i/o exception, change exception message
			throw new PersistorException("1", e);
		}
	}

	private InnerBPlusTreeNode<K, V> deserializeInnerNode(byte[] bytes) throws PersistorException {
		try {
			DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes));
			input.readByte();

			FileBPlusTreeNodeReference<K, V> parentReference = readFileReference(input);

			int keyCount = input.readInt();

			ArrayList<K> keys = new ArrayList<K>(keyCount);
			for (int i = 0; i < keyCount; i++) {
				keys.add(readKey(input));
			}

			int referenceCount = (keyCount > 0) ? keyCount + 1 : 0;
			List<BPlusTreeNodeReference<K, V>> references = new ArrayList<BPlusTreeNodeReference<K, V>>(referenceCount);
			for (int i = 0; i < referenceCount; i++) {
				references.add(readFileReference(input));
			}

			input.close();

			InnerBPlusTreeNode<K, V> node = new InnerBPlusTreeNode<K, V>(tree, parentReference);
			node.setKeys(keys);
			node.setReferences(references);
			return node;
		} catch (IOException e) {
			e.printStackTrace();
			// TODO: change exception message
			throw new PersistorException("4", e);
		}
	}

	private byte[] serializeLeafNode(LeafBPlusTreeNode<K, V> node) throws PersistorException {
		try {
			ByteArrayOutputStream baos = new ByteArrayOutputStream();
			DataOutputStream output = new DataOutputStream(baos);

			output.writeByte(0);

			writeFileReference(output, (FileBPlusTreeNodeReference<K, V>)node.getParentReference());
			
			writeFileReference(output, (FileBPlusTreeNodeReference<K, V>)node.getPreviousReference());

			writeFileReference(output, (FileBPlusTreeNodeReference<K, V>)node.getNextReference());

			Map<K, V> data = node.getData();
			output.writeInt(data.size());

			for (K key : data.keySet()) {
				writeKey(output, key);
				writeValue(output, data.get(key));
			}

			output.close();
			return baos.toByteArray();
		} catch (IOException e) {
			e.printStackTrace();
			// TODO: change exception message
			throw new PersistorException("1");
		}
	}

	private LeafBPlusTreeNode<K, V> deserializeLeafNode(byte[] bytes) throws PersistorException {
		try {
			DataInputStream input = new DataInputStream(new ByteArrayInputStream(bytes));
			input.readByte();

			FileBPlusTreeNodeReference<K, V> parentReference = readFileReference(input);

			FileBPlusTreeNodeReference<K, V> previousReference = readFileReference(input);

			FileBPlusTreeNodeReference<K, V> nextReference = readFileReference(input);

			int size = input.readInt();
			Map<K, V> data = new TreeMap<K, V>();
			for (int i = 0; i < size; i++) {
				K key = readKey(input);
				V value = readValue(input);
				data.put(key, value);
			}

			input.close();

			LeafBPlusTreeNode<K, V> node = new LeafBPlusTreeNode<K, V>(tree, parentReference);
			node.setPreviousReference(previousReference);
			node.setNextReference(nextReference);
			node.setData(data);
			return node;
		} catch (IOException e) {
			e.printStackTrace();
			// TODO: change exception message
			throw new PersistorException("1", e);
		}
	}

	private void writeFileReference(DataOutputStream output, FileBPlusTreeNodeReference<K, V> reference) throws PersistorException {
		try {
			int block = (reference == null) ? -1 : reference.getBlock();
			int record = (reference == null) ? -1 : reference.getRecord();
			output.writeInt(block);
			output.writeInt(record);
		} catch (IOException e) {
			e.printStackTrace();
			//TODO: change exception message
			throw new PersistorException("1", e);
		}
	}

	private FileBPlusTreeNodeReference<K, V> readFileReference(DataInputStream input) throws PersistorException {
		try {
			int block = input.readInt();
			int record = input.readInt();
			if (block == -1 && record == -1)
				return null;
			else
				return new FileBPlusTreeNodeReference<K, V>(tree, block, record);
		} catch (IOException e) {
			e.printStackTrace();
			//TODO: change exception message
			throw new PersistorException("1", e); 
		}
	}

	private void writeKey(DataOutputStream output, K key) throws PersistorException {
		try {
			byte[] bytes = tree.getKeyPersistor().serialize(key);
			output.writeInt(bytes.length);
			output.write(bytes);
		} catch (IOException e) {
			e.printStackTrace();
			// TODO: change exception message
			throw new PersistorException("1", e);
		} catch (PersistorException e) {
			e.printStackTrace();
			// TODO: change exception message
			throw new PersistorException("2", e);
		}
	}

	private K readKey(DataInputStream input) throws PersistorException {
		try {
			int size = input.readInt();
			byte[] bytes = new byte[size];
			input.read(bytes);
			return this.getTree().getKeyPersistor().deserialize(bytes);
		} catch (IOException e) {
			e.printStackTrace();
			// TODO: change exception message
			throw new PersistorException("1", e);
		} catch (PersistorException e) {
			e.printStackTrace();
			// TODO: change exception message
			throw new PersistorException("2", e);
		}
	}

	private void writeValue(DataOutputStream output, V value) throws PersistorException {
		try {
			byte[] bytes = tree.getValuePersistor().serialize(value);
			output.writeInt(bytes.length);
			output.write(bytes);
		} catch (IOException e) {
			e.printStackTrace();
			// TODO: change exception message
			throw new PersistorException("1", e);
		}
	}

	private V readValue(DataInputStream input) throws PersistorException {
		try {
			int size = input.readInt();
			byte[] bytes = new byte[size];
			input.read(bytes);
			return this.getTree().getValuePersistor().deserialize(bytes);
		} catch (IOException e) {
			e.printStackTrace();
			// TODO: change exception message
			throw new PersistorException("1", e);
		} catch (PersistorException e) {
			e.printStackTrace();
			// TODO: change exception message
			throw new PersistorException("2", e);
		}
	}
	
	private FileBPlusTree<K, V> getTree() {
		return this.tree;
	}

	private void setTree(FileBPlusTree<K, V> tree) {
		this.tree = tree;
	}
}
